package com.kamistoat.rocketmqlearning.learnFromRoacketMQ;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;

public class MmapLearning {
    private static HashMap<String, MyMappedFile> myMappedFiles = new HashMap<>();
    private static final int fileSize = 1024;

    public static void main(String[] args) {
        createMyMappedFile();
        Thread reputThread = new Thread(new RePutService());
        reputThread.start();
        System.out.println("all started!!");
        doAppendMsg("First Message...");
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        doAppendMsg("Second Message...");
    }

    /**
     * 创建MyMappedFile的mmap映射
     */
    private static void createMyMappedFile() {
        try {
            // 创建普通File
            File file = new File("/home/kamistoat/ubuntu/RocketMQLearning/src/main/java/com/kamistoat/rocketmqlearning/learnFromRoacketMQ/mymappedfile.jj");
            // 获取FileChannel
            FileChannel fileChannel = new RandomAccessFile(file, "rw").getChannel();
            // mmap映射。注意mmap只能做区间映射。要么明确文件区间，要么固定文件大小，每次都将整个文件映射到mmap
            MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, fileSize);
            // 关闭FileChannel
            fileChannel.close();

            // 读取首个记录的长度，查看mmap内存中是否已经存在旧记录。
            int msgLen = mappedByteBuffer.getInt(0);
            int readableOffset = 0;
            // 如果mmap中存在旧记录，要一直跳到最后。
            while (msgLen != 0) {
                readableOffset = readableOffset + 4 + msgLen;
                msgLen = mappedByteBuffer.getInt(readableOffset);
            }

            MyMappedFile myMappedFile = new MyMappedFile(readableOffset, mappedByteBuffer);
            myMappedFiles.put("mymappedfile", myMappedFile);
            System.out.println("MyMappedFile created successfully!");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 向mmap中追加内容，然后将内存刷盘
     * @param msg
     */
    private static void doAppendMsg(String msg) {
        // 获取MyMappedFile
        MyMappedFile mymappedfile = myMappedFiles.get("mymappedfile");

        MappedByteBuffer mappedByteBuffer = mymappedfile.getMappedByteBuffer();
        int readableOffset = mymappedfile.getReadableOffset();

        byte[] msg_append = ("Append '" + msg + "', Time at: " + new Date().toString()).getBytes(StandardCharsets.UTF_8);
        int msg_append_len = msg_append.length;
        if (readableOffset + msg_append_len > fileSize) {
            System.out.println("MyMappedFile has no more space for appending message!");
        } else {
            mappedByteBuffer.position(readableOffset);
            mappedByteBuffer.putInt(msg_append.length);
            mappedByteBuffer.put(msg_append);
            readableOffset = readableOffset + 4 + msg_append.length;
            mymappedfile.setReadableOffset(readableOffset);

            // 强制将mmap内存刷新到磁盘上。实际开发中，这一步如果存在多线程，要加锁做同步阻塞刷盘。
            mappedByteBuffer.force();

            System.out.println("MyMappedFile append successfully!");
            myMappedFiles.put("mymappedfile", mymappedfile);
        }
    }

    /**
     * 异步的reput操作
     */
    private static class RePutService implements Runnable {
        private int rePutOffset;

        /**
         * 开启异步后，死循环执行。
         */
        @Override
        public void run() {
            rePutOffset = 0;
            while (true) {
                try {
                    Thread.sleep(3000);
                    doReput();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

        private void doReput() {
            MyMappedFile mymappedfile = myMappedFiles.get("mymappedfile");
            int readableOffset = mymappedfile.getReadableOffset();
            MappedByteBuffer mappedByteBuffer = mymappedfile.getMappedByteBuffer();
            System.out.format("Try to reput %d byte from %d to %d offset in mymappedfile... \n",
                    readableOffset - rePutOffset,
                    rePutOffset,
                    readableOffset);

            while(rePutOffset < readableOffset){
                mappedByteBuffer.position(rePutOffset);
                int msgLen = mappedByteBuffer.getInt();
                ByteBuffer tempBuffer = mappedByteBuffer.slice();
                tempBuffer.limit(msgLen);
                rePutOffset = rePutOffset + 4 + msgLen;
                byte[] bytes = new byte[msgLen];
                tempBuffer.get(bytes);
                System.out.println("Reput msg: '" + new String(bytes) + "'");
            }
        }
    }

    private static class MyMappedFile {
        private int readableOffset;
        private MappedByteBuffer mappedByteBuffer;

        public MyMappedFile() {
        }

        public MyMappedFile(int readableOffset, MappedByteBuffer mappedByteBuffer) {
            this.readableOffset = readableOffset;
            this.mappedByteBuffer = mappedByteBuffer;
        }

        public int getReadableOffset() {
            return readableOffset;
        }

        public void setReadableOffset(int readableOffset) {
            this.readableOffset = readableOffset;
        }

        public MappedByteBuffer getMappedByteBuffer() {
            return mappedByteBuffer;
        }

        public void setMappedByteBuffer(MappedByteBuffer mappedByteBuffer) {
            this.mappedByteBuffer = mappedByteBuffer;
        }
    }
}
